import warnings
from pathlib import Path
from typing import Any, Callable, Literal

import pandas as pd
from pydantic import BaseModel, Field

from ..dataclass.results import SingleResult, UnitResult
from ..dataclass.tables import TableCollection, TableData
from ..dataclass.terminologies import DataTypes
from ..pipeline.nameSpace import load_local_function_provider
from ..pipeline.pipeData import (
    ArrayAttributeSchema,
    BooleanAttributeSchema,
    MultiResultsDict,
    NestedRangeModel,
    RangeModel,
    ResultsDict,
    StringAttributeSchema,
    UIAttributeSchema,
)
from ..pipeline.pipeline import (
    PipeModule,
    Port,
    PortReference,
    module_decorator,
    status_manage,
)
from ..pipeline.portTypes import PortType, PortTypeHint


@status_manage
class TableSeriesSelector(PipeModule):
    """Select a series of tables from a TableData object."""

    def __init__(
        self,
        mname: str | None = None,
        auto_run: bool = True,
        table_collection: PortTypeHint.TableCollection | None = None,
        select_field: tuple[str, str] | None = None,
        names_type: Literal["name", "title"] = "title",
    ) -> None:
        """Initialize TableSeriesSelector object.

        Parameters
        ----------
        table_collection : TableCollection
            The table collection to select from.

        select_field : tuple[str, str]
            The field to select from. The first element is the table name, the second element is the field name.

        names_type : Literal["name", "title"]
            The type of the table name.
            If it's `name`, the name will be used which is used in the database.
            If it's `title`, the title will be used which is user-friendly.
        """
        super().__init__(mname, auto_run)

        self._ports_in = {
            "InputTables": Port(
                PortType.TableCollection,
                data=table_collection,
                pdoc="The table collection to select from.",
            )
        }
        self._ports_out = {
            "OutputTableSeries": Port(
                PortType.TableSeries, pdoc="The selected table series."
            )
        }

        self.select_field = select_field
        self.names_type = names_type

    @property
    def InputTables(self) -> None:
        raise AttributeError("Property 'InputTables' is read-only.")

    @InputTables.setter
    def InputTableCollection(self, value: PortTypeHint.TableCollection) -> None:
        self._ports_in["InputTables"].data = value

    @property
    def OutputTableSeries(self) -> PortTypeHint.TableSeries | None:
        return self._ports_out["OutputTableSeries"].data

    def set_cal_params(self, reset: bool = False) -> dict[str, RangeModel]:
        return {
            "select_field": RangeModel(
                vtype="tuple", tuple_len=2, default=self.select_field, title="选择字段"
            ),
            "names_type": RangeModel(
                vtype="str",
                default=self.names_type,
                choices=("name", "title"),
                title="表名和字段名类型",
            ),
        }

    def execute(self) -> PortTypeHint.TableSeries | None:
        tables: PortTypeHint.TableCollection | None = self._ports_in["InputTables"].data

        if tables is None or self.select_field is None:
            self._ports_out["OutputTableSeries"].data = None
            return None

        table_name, field_name = self.select_field
        table = tables[table_name]

        if self.names_type == "title":
            field_name = table.get_field_metadata(field_name).name

        table_series = table[field_name]
        self._ports_out["OutputTableSeries"].data = table_series
        return table_series


@module_decorator()
class TableSelector(PipeModule):
    """Select a table from a TableCollection."""

    InputTables: PortReference[PortTypeHint.TableCollection]
    OutputTable: PortReference[PortTypeHint.TableData]

    _port_docs = {
        "InputTables": "The table collection to select from.",
        "OutputTable": "The selected table.",
    }

    def __init__(
        self,
        mname: str | None = None,
        auto_run: bool = True,
        tables: PortTypeHint.TableCollection | None = None,
        table_name: str | None = None,
        table_idx: int | None = None,
    ) -> None:
        """Initialize TableSelector object.

        Parameters
        ----------
        tables : TableCollection
            The table collection to select from.

        table_name : str, default: None
            The name or title of the table to select.
            If not None, 'table_idx' will be ignored.

        table_idx : int, default: None
            The index of table in the table collection.
        """
        super().__init__(mname, auto_run)

        if tables is not None:
            self.InputTables = tables

        self.table_name = table_name
        self.table_idx = table_idx

    def update_ui_schema(self, reset: bool = False) -> dict[str, UIAttributeSchema]:
        tables: PortTypeHint.TableCollection | None = self._ports_in["InputTables"].data

        if tables is None or len(tables) == 0:
            table_name_selections = None
        else:
            table_name_selections = tables.table_titles

        return {
            "table_name": StringAttributeSchema(
                title="表名称",
                selections=table_name_selections,
            )
        }

    def set_cal_params(self, reset: bool = True) -> dict[str, RangeModel]:
        tables: PortTypeHint.TableCollection | None = self._ports_in["InputTables"].data

        if tables is None or len(tables) == 0:
            table_name_choices = None
        else:
            table_name_choices = tables.table_titles
            if reset:
                self.table_name = table_name_choices[0]

        return {
            "table_name": RangeModel(
                vtype="str",
                default=self.table_name,
                choices=table_name_choices,
                title="表名称",
            )
        }

    def execute(self) -> PortTypeHint.TableData | None:
        tables: PortTypeHint.TableCollection | None = self._ports_in["InputTables"].data

        if not tables:
            self._ports_out["OutputTable"].data = None
            return None

        if self.table_name:
            selected_table = tables[self.table_name]
        elif self.table_idx is not None:
            selected_table = tables[self.table_idx]
        else:
            selected_table = None

        self._ports_out["OutputTable"].data = selected_table
        return selected_table


@module_decorator()
class TableCollectionSelector(PipeModule):
    """Select tables or remove tables from a TableCollection."""

    InputTables: PortReference[PortTypeHint.TableCollection]
    OutputTables: PortReference[PortTypeHint.TableCollection]

    _port_docs = {
        "InputTables": "The table collection to select from.",
        "OutputTables": "The selected table collection.",
    }

    def __init__(
        self,
        mname: str | None = None,
        auto_run: bool = True,
        tables: PortTypeHint.TableCollection | None = None,
        table_names: list[str] | None = None,
        table_idxs: list[int] | None = None,
        operation: Literal["select", "remove"] = "select",
    ) -> None:
        """Initialize TableCollectionSelector object.

        Parameters
        ----------
        tables : TableCollection
            The table collection to select from.

        table_names : list[str] | None
            The names or titles of the tables to select or remove.
            If the table_name is not in the table collection, it will be ignored.
            If not None, 'table_idxs' will be ignored.

        table_idxs : list[int] | None
            The indices of the tables to select or remove.
            If the table_idx is out of range, it will be ignored.

        operation : Literal["select", "remove"]
            The operation to perform on the table collection.
            If it's 'select', the tables will be selected from the table collection.
            If it's 'remove', the tables will be removed from the table collection.

        Note
        ----
        A new table collections will be created so that the original table collection will not be modified.
        If table_names and table_idxs are both None:
            If operation is `select`, None will be returned.
            If operation is `remove`, the original table collection will be returned.
        """
        super().__init__(mname, auto_run)

        if tables is not None:
            self.InputTables = tables

        self.table_names = table_names
        self.table_idxs = table_idxs
        self.operation = operation

    def execute(self) -> PortTypeHint.TableCollection | None:
        """Execute the table selection or removal operation.

        Returns
        -------
        TableCollection or None
            The resulting table collection after selection/removal operation.
        """
        tables: PortTypeHint.TableCollection | None = self._ports_in["InputTables"].data

        if tables is None:
            self._ports_out["OutputTables"].data = None
            return None

        if not self.table_names and not self.table_idxs:
            if self.operation == "select":
                self._ports_out["OutputTables"].data = None
                return None
            elif self.operation == "remove":
                self._ports_out["OutputTables"].data = tables
                return tables

        # Determine which tables to operate on
        target_tables = []

        if self.table_names:
            # Use table names/titles (priority over indices)
            for name_or_title in self.table_names:
                table = tables.get_table(name_or_title)
                if table is not None:
                    target_tables.append(table)
        else:
            # Use table indices
            for idx in self.table_idxs:
                try:
                    table = tables.get_table_by_index(idx)
                    target_tables.append(table)
                except IndexError:
                    # If any index is out of range, continue
                    continue

        if self.operation == "select":
            output_tables = TableCollection(
                name=tables.name,
                title=tables.title,
                description=tables.description,
            )
            for table in target_tables:
                output_tables.add_table(table.copy())
        elif self.operation == "remove":
            # Create a new table collection (copy original to avoid modifying it)
            output_tables = tables.copy()
            for table in target_tables:
                output_tables.remove_table(table.name)

        self._ports_out["OutputTables"].data = output_tables
        return output_tables


class _BaseTablesRowFilter(PipeModule):
    """Base class for filtering rows of tables based on conditions."""

    # Define supported filter operation types
    FILTER_OPERATIONS = {
        "eq": lambda x, y: x == y,
        "ne": lambda x, y: x != y,
        "gt": lambda x, y: x > y,
        "lt": lambda x, y: x < y,
        "ge": lambda x, y: x >= y,
        "le": lambda x, y: x <= y,
        "in": lambda x, y: x in y,
        "not_in": lambda x, y: x not in y,
        "contains": lambda x, y: y in x if isinstance(x, str) else False,
        "starts_with": lambda x, y: x.startswith(y) if isinstance(x, str) else False,
        "ends_with": lambda x, y: x.endswith(y) if isinstance(x, str) else False,
        "is_null": lambda x, y: pd.isna(x),
        "is_not_null": lambda x, y: pd.notna(x),
    }

    def __init__(
        self,
        mname: str | None = None,
        auto_run: bool = True,
        tables: PortTypeHint.TableCollection | None = None,
        filter_by: str | None = None,
        filter_operation: (
            Literal[
                "eq",
                "ne",
                "gt",
                "lt",
                "ge",
                "le",
                "in",
                "not_in",
                "contains",
                "starts_with",
                "ends_with",
                "is_null",
                "is_not_null",
            ]
            | None
        ) = None,
        filter_value: Any = None,
        min_items: int = 0,
        max_items: int | None = None,
        filter_by_read_only: bool = False,
        filter_operation_read_only: bool = False,
        fiter_by_visible: bool = True,
        filter_operation_visible: bool = True,
        filter_value_type: Literal["str", "int", "float", "list", "bool"] | None = None,
        filter_value_choices: Literal["by_uniques"] | None = None,
    ) -> None:
        """Initialize base table row filter object.

        Parameters
        ----------
        tables : TableCollection
            The table collection to filter.

        filter_by : str
            The column name or title to apply the filter on.

        filter_operation : str, optional
            Predefined filter operation to use ('eq', 'ne', 'gt', 'lt', 'ge', 'le', etc.)
            Check FILTER_OPERATIONS for all supported operations.

        filter_value : Any, optional
            The value to compare against when using filter_operation.

        min_items : int, default: 0
            The minimum number of filters.

        max_items : int | None, default: None
            The maximum number of filters.
            If None, no limit.

        filter_by_read_only : bool, default: False
            Whether the filter_by is read-only.
            If it's True, the filter_by cannot be changed by the user.

        filter_operation_read_only : bool, default: False
            Whether the filter_operation is read-only.
            If it's True, the filter_operation cannot be changed by the user.

        fiter_by_visible : bool, default: True
            Whether the filter_by is visible.
            If it's False, the filter_by will not be shown in the UI.

        filter_operation_visible : bool, default: True
            Whether the filter_operation is visible.
            If it's False, the filter_operation will not be shown in the UI.

        filter_value_type : Literal["str", "int", "float", "list", "bool"] | None, default: None
            The type of the filter value. It's used in RangeModel.
            If it's None, the filter value will be a string.

        filter_value_choices : Literal["by_uniques"] | None, default: None
            The choices of the filter value. It's used in RangeModel.
            If it's "by_uniques", the filter value will be the unique values of the filter_by column.
            If it's None, there is no choices for the filter value.
        """
        super().__init__(mname, auto_run)

        self._ports_in = {
            "InputTables": Port(
                PortType.TableCollection,
                data=tables,
                pdoc="The table collection to filter.",
            )
        }
        self._ports_out = {
            "OutputTables": Port(
                PortType.TableCollection, pdoc="The filtered table collection."
            )
        }

        # filters example - [{"filter_by": str, "filter_operation": str, "filter_value": Any}]
        self.filters = []
        if filter_by and filter_operation and filter_value:
            self.add_filter(
                filter_by=filter_by,
                filter_operation=filter_operation,
                filter_value=filter_value,
            )

        self.min_items = min_items
        self.max_items = max_items
        self.filter_by_read_only = filter_by_read_only
        self.filter_operation_read_only = filter_operation_read_only
        self.fiter_by_visible = fiter_by_visible
        self.filter_operation_visible = filter_operation_visible
        self.filter_value_type = filter_value_type
        self.filter_value_choices = filter_value_choices

    @property
    def InputTables(self) -> None:
        raise AttributeError("Property 'InputTables' is write-only.")

    @InputTables.setter
    def InputTables(self, value: PortTypeHint.TableCollection | None) -> None:
        self._ports_in["InputTables"].data = value

    @property
    def OutputTables(self) -> PortTypeHint.TableCollection | None:
        return self._ports_out["OutputTables"].data

    def add_filter(
        self,
        filter_by: str,
        filter_operation: (
            Literal[
                "eq",
                "ne",
                "gt",
                "lt",
                "ge",
                "le",
                "in",
                "not_in",
                "contains",
                "starts_with",
                "ends_with",
                "is_null",
                "is_not_null",
            ]
            | None
        ) = None,
        filter_value: Any = None,
    ) -> None:
        """Add additional filter to the module.

        Parameters
        ----------
        filter_by : str
            The column name or title to apply the filter on.

        filter_operation : str, optional
            Predefined filter operation to use ('eq', 'ne', 'gt', 'lt', 'ge', 'le', etc.)
            Check FILTER_OPERATIONS for all supported operations.

        filter_value : Any, optional
            The value to compare against when using filter_operation.
            If it's None, the common filter value will be used.
        """
        self.filters.append(
            {
                "filter_by": filter_by,
                "filter_operation": filter_operation,
                "filter_value": filter_value,
            }
        )

    def clear_filters(self) -> None:
        """Clear all filters."""
        self.filters = []

    # def add_custom_filter(self, filter_condition: Callable) -> None:
    #     """Add a custom filter to the module.

    #     Parameters
    #     ----------
    #     filter_condition : callable
    #         A custom function that takes a value and returns a boolean.
    #         This is not saved in the pipeline - for custom filtering logic.
    #     """
    #     self._customer_filter = filter_condition

    def _get_filter_function(self, filter_config: dict) -> callable:
        """Return the appropriate filter function based on configuration."""
        filter_operation = filter_config["filter_operation"]
        filter_value = filter_config["filter_value"]

        if filter_operation in self.FILTER_OPERATIONS:
            operation_func = self.FILTER_OPERATIONS[filter_operation]
            return lambda x: operation_func(x, filter_value)
        else:
            raise ValueError(
                f"The filter operation '{filter_operation}' is not supported."
            )


@status_manage
class RelatedTablesRowFilter(_BaseTablesRowFilter):
    """Filter rows of a series related tables based on a condition."""

    def __init__(
        self,
        mname: str | None = None,
        auto_run: bool = True,
        tables: PortTypeHint.TableCollection | None = None,
        filter_by: str | None = None,
        filter_operation: (
            Literal[
                "eq",
                "ne",
                "gt",
                "lt",
                "ge",
                "le",
                "in",
                "not_in",
                "contains",
                "starts_with",
                "ends_with",
            ]
            | None
        ) = None,
        filter_value: Any = None,
        min_items: int = 0,
        max_items: int | None = None,
        filter_by_read_only: bool = False,
        filter_operation_read_only: bool = False,
        fiter_by_visible: bool = True,
        filter_operation_visible: bool = True,
        filter_value_type: Literal["str", "int", "float", "list", "bool"] | None = None,
        filter_value_choices: Literal["by_uniques"] | None = None,
        main_table: str | None = None,
        sub_tables: list[str] | None = None,
        main_key: str | None = None,
        local_functions_path: str | None = None,
        local_functions_name: str | None = None,
    ) -> None:
        """Initialize TablesRowFilter object.

        Parameters
        ----------
        tables : TableCollection
            The table collection to filter.

        filter_by : str
            The column name or title to apply the filter on.

        filter_operation : str, optional
            Predefined filter operation to use ('eq', 'ne', 'gt', 'lt', 'ge', 'le', etc.)
            Check FILTER_OPERATIONS for all supported operations.

        filter_value : Any, optional
            The value to compare against when using filter_operation.

        min_items : int, default: 0
            The minimum number of filters.

        max_items : int | None, default: None
            The maximum number of filters.
            If None, no limit.

        filter_by_read_only : bool, default: False
            Whether the filter_by is read-only.
            If it's True, the filter_by cannot be changed by the user.

        filter_operation_read_only : bool, default: False
            Whether the filter_operation is read-only.
            If it's True, the filter_operation cannot be changed by the user.

        fiter_by_visible : bool, default: True
            Whether the filter_by is visible.
            If it's False, the filter_by will not be shown in the UI.

        filter_operation_visible : bool, default: True
            Whether the filter_operation is visible.
            If it's False, the filter_operation will not be shown in the UI.

        filter_value_type : Literal["str", "int", "float", "list", "bool"] | None, default: None
            The type of the filter value. It's used in RangeModel.
            If it's None, the filter value will be a string.

        filter_value_choices : Literal["by_uniques"] | None, default: None
            The choices of the filter value. It's used in RangeModel.
            If it's None, the filter value will be a string.

        main_table : str
            The name of the main table to apply the filter on.

        sub_tables : list[str], optional
            List of other tables to filter based on filtered main table.

        main_key : str, optional
            The key column in the main table to use for joining with sub tables.

        local_functions_path : str, optional
            The path to the local functions file.
            If it's not provided, the module will try to use the local functions file path in the pipeline.

        local_functions_name : str, optional
            The name of the local function that defines the function about how to get the filter value range model.
            The module will use the path of the local functions file to load the class.

        Note
        ----
        Only the main table and the specified sub tables will be included in the output table collection, othere tables will be ignored.
        """
        super().__init__(
            mname,
            auto_run,
            tables,
            filter_by,
            filter_operation,
            filter_value,
            min_items,
            max_items,
            filter_by_read_only,
            filter_operation_read_only,
            fiter_by_visible,
            filter_operation_visible,
            filter_value_type,
            filter_value_choices,
        )

        self.main_table = main_table
        self.sub_tables = sub_tables
        self.main_key = main_key
        self.local_functions_path = local_functions_path
        self.local_functions_name = local_functions_name

    def set_cal_params(
        self, reset: bool = True
    ) -> dict[str, RangeModel | NestedRangeModel]:
        """Define parameters for the module."""
        tables: PortTypeHint.TableCollection | None = self._ports_in["InputTables"].data

        main_table_choices = None
        sub_tables_choices = None
        main_key_choices = None
        filter_by_choices = None
        filter_value_choices = None
        filter_value_depends = None

        if tables and len(tables) > 0:
            # Get and set main table
            main_table_choices = tables.table_titles
            if not self.main_table and reset:
                self.main_table = main_table_choices[0]
            # Get and set sub tables
            if self.main_table:
                sub_tables_choices = [
                    x for x in tables.table_titles if x != self.main_table
                ]
                if not self.sub_tables and reset:
                    self.sub_tables = sub_tables_choices
            # Get and set main key
            if self.main_table:
                main_key_choices = tables[self.main_table].field_titles
                if not self.main_key and reset:
                    self.main_key = main_key_choices[0]
            # Get and set filter by
            if self.main_table:
                filter_by_choices = tables[self.main_table].field_titles
                # Reset the filters when it's empty
                if not self.filters and reset:
                    self.filters.append(
                        {
                            "filter_by": filter_by_choices[0],
                            "filter_operation": "eq",
                            "filter_value": None,
                        }
                    )
                # Get the choices of the filter value for case 'by_uniques'
                if self.filter_value_choices == "by_uniques":
                    filter_value_depends = "filter_by"
                    filter_value_choices = {}
                    if self.filters:
                        existed_filter_bys = []
                        for filter in self.filters:
                            filter_by = filter["filter_by"]
                            if filter_by and filter_by not in existed_filter_bys:
                                existed_filter_bys.append(filter_by)
                                filter_value_choices[filter_by] = set()
                                main_table = tables[self.main_table]
                                if (
                                    filter_by in main_table.field_titles
                                    or filter_by in main_table.columns
                                ):
                                    filter_value_choices[filter_by] = (
                                        main_table[filter_by].unique().tolist()
                                    )
                    if not filter_value_choices:
                        filter_value_choices = None

        filter_ops = list(self.FILTER_OPERATIONS.keys())
        filter_by_range = RangeModel(
            vtype="str",
            title="主表过滤字段",
            choices=filter_by_choices,
            depends_on=["main_table"],
            read_only=self.filter_by_read_only,
            visible=self.fiter_by_visible,
        )
        filter_operation_range = RangeModel(
            vtype="str",
            title="过滤条件",
            choices=filter_ops,
            choices_label=[
                "等于",
                "不等于",
                "大于",
                "小于",
                "大于等于",
                "小于等于",
                "在...中",
                "不在...中",
                "包含",
                "以...开始",
                "以...结束",
            ],
            read_only=self.filter_operation_read_only,
            visible=self.filter_operation_visible,
        )
        filter_value_range = RangeModel(
            vtype=self.filter_value_type or "str",
            title="过滤值",
            choices=filter_value_choices,
            depends_on=filter_value_depends,
        )

        # Get filter value range model by local function
        local_function_exists = False
        if self.local_functions_path:
            local_function_exists = True
            local_functions_path = self.local_functions_path
        # Try to get the local function path from the pipeline if self.local_functions_path is not provided
        elif self.pipeline and self.pipeline.local_functions_path:
            local_function_exists = True
            local_functions_path = self.pipeline.local_functions_path

        if local_function_exists:
            filter_value_range_class = load_local_function_provider(
                local_functions_path, self.local_functions_name
            )
            # Get the function context of the class
            function_context = filter_value_range_class.context
            for key in function_context.keys():
                if key in self.keys()["input_ports"]:
                    value = self._ports_in[key].data
                elif key in self.keys()["output_ports"]:
                    value = self._ports_out[key].data
                else:
                    try:
                        value = getattr(self, key)
                    except AttributeError:
                        raise ValueError(
                            f"The attribute '{key}' is not found in the current module '{self.mname}'."
                        )
                function_context[key] = value

            filter_value_range_class.update_context(function_context)

            filter_value_range = filter_value_range_class.run_function()
            if not isinstance(filter_value_range, RangeModel):
                raise ValueError(
                    f"The result of the function '{self.local_functions_name}' is not a RangeModel."
                )

        values_range = {
            "main_table": RangeModel(
                vtype="str",
                default=self.main_table,
                title="主表名称",
                choices=main_table_choices,
            ),
            "sub_tables": RangeModel(
                vtype="list",
                list_type="str",
                default=self.sub_tables,
                title="子表名称",
                choices=sub_tables_choices,
                depends_on=["main_table"],
            ),
            "main_key": RangeModel(
                vtype="str",
                default=self.main_key,
                title="主表主键",
                choices=main_key_choices,
                depends_on=["main_table"],
            ),
            "filters": NestedRangeModel(
                vtype="list_dict",
                title="条件过滤器",
                fields={
                    "filter_by": filter_by_range,
                    "filter_operation": filter_operation_range,
                    "filter_value": filter_value_range,
                },
                default=self.filters,
                min_max_items=(self.min_items, self.max_items),
            ),
        }
        return values_range

    def execute(self) -> PortTypeHint.TableCollection | None:
        """Execute the filter on the table collection."""
        tables: PortTypeHint.TableCollection | None = self._ports_in["InputTables"].data

        if tables is None or self.main_table is None or not self.filters:
            self._ports_out["OutputTables"].data = None
            return None

        # Create a new table collection to store the filtered tables
        output_tables = TableCollection()
        # Get the main table
        main_table = tables[self.main_table]
        # Apply filters
        table_mask = None
        for filter_config in self.filters:
            filter_by = filter_config["filter_by"]

            # Skip if the main table doesn't have this field
            if (filter_by not in main_table.field_titles) and (
                filter_by not in main_table.columns
            ):
                continue

            # Get the filter function for this condition
            filter_func = self._get_filter_function(filter_config)
            # Apply the filter
            current_mask = main_table[filter_by].apply(filter_func)
            # Combine with previous filters (AND operation)
            if table_mask is None:
                table_mask = current_mask
            else:
                table_mask = table_mask & current_mask

        # Add the filtered main table to the output table collection
        filtered_main_table = main_table[table_mask]
        output_tables.add_table(filtered_main_table)

        # Filter sub tables if specified
        if self.sub_tables and self.main_key:
            # Get the filtered key values from main table
            filtered_keys = set(filtered_main_table[self.main_key].tolist())

            # Filter each sub table based on the keys
            for sub_table_name in self.sub_tables:
                if sub_table_name in tables and sub_table_name != self.main_table:
                    sub_table = tables[sub_table_name]
                    filtered_sub_table = sub_table[
                        sub_table[self.main_key].isin(filtered_keys)
                    ].copy()
                    # Add the filtered sub table to the output table collection
                    output_tables.add_table(filtered_sub_table)

        self._ports_out["OutputTables"].data = output_tables
        return output_tables


@status_manage
class TablesRowFilter(_BaseTablesRowFilter):
    """Filter rows of each table from a table collection or a single table based on a condition."""

    def __init__(
        self,
        mname: str | None = "TablesRowFilter",
        auto_run: bool = True,
        tables: PortTypeHint.TableCollection | PortTypeHint.TableData | None = None,
        filter_by: str = None,
        filter_operation: (
            Literal[
                "eq",
                "ne",
                "gt",
                "lt",
                "ge",
                "le",
                "in",
                "not_in",
                "contains",
                "starts_with",
                "ends_with",
                "is_null",
                "is_not_null",
            ]
            | None
        ) = None,
        filter_value: Any = None,
        min_items: int = 0,
        max_items: int | None = None,
        filter_by_read_only: bool = False,
        filter_operation_read_only: bool = False,
        fiter_by_visible: bool = True,
        filter_operation_visible: bool = True,
        filter_value_type: Literal["str", "int", "float", "list", "bool"] | None = None,
        filter_value_choices: Literal["by_uniques"] | None = None,
        module_attributes_map: (
            dict[str, str] | dict[str, tuple[str, str] | list[str]] | None
        ) = None,
        input_attributes_port_required: bool = False,
    ) -> None:
        """Initialize TablesRowFilter object.

        Parameters
        ----------
        tables : TableCollection | TableData
            The table collection or single table to filter.

        filter_by : str
            The column name or title to apply the filter on.

        filter_operation : str, optional
            Predefined filter operation to use ('eq', 'ne', 'gt', 'lt', 'ge', 'le', etc.)
            Check FILTER_OPERATIONS for all supported operations.

        filter_value : Any, optional
            The value to compare against when using filter_operation.

        filter_value_type : Literal["str", "int", "float", "list", "bool"] | None, default: None
            The type of the filter value. It's used in RangeModel.
            If it's None, the filter value will be a string.

        module_attributes_map : dict[str, str] | dict[str, tuple[str, str] | list[str]] | None
            The map of the other module's attributes to the current module's attributes.

        input_attributes_port_required : bool
            Whether the "InputAttributes" port data is required.

        Note
        ----
        - When input is a TableCollection: All tables with field `filter_by` will be included in the output table collection.
        - When input is a single TableData: The filtered table will be returned as a single TableData object.
        - Multiple filter conditions can be added using the add_filter method.
        """
        super().__init__(
            mname,
            auto_run,
            tables,
            filter_by,
            filter_operation,
            filter_value,
            min_items,
            max_items,
            filter_by_read_only,
            filter_operation_read_only,
            fiter_by_visible,
            filter_operation_visible,
            filter_value_type,
            filter_value_choices,
        )

        # Overwrite the input and output ports to support both TableCollection and TableData
        self._ports_in = {
            "InputTables": Port(
                ptype=[PortType.TableCollection, PortType.TableData],
                data=tables,
                pdoc="The table collection or single table to filter.",
            )
        }
        self._ports_out = {
            "OutputTables": Port(
                ptype=[PortType.TableCollection, PortType.TableData],
                pdoc="The filtered table collection or single table.",
            ),
            "OutputTablesRowMask": Port(
                ptype=[PortType.TablesRowMask, PortType.TableRowMask],
                pdoc="The row mask of the filtered tables.",
            ),
        }

        self.module_attributes_map = module_attributes_map
        self.input_attributes_port_required = input_attributes_port_required
        self._disable_input_attributes = True

    @property
    def InputTables(self) -> None:
        raise AttributeError("Property 'InputTables' is write-only.")

    @InputTables.setter
    def InputTables(
        self, value: PortTypeHint.TableCollection | PortTypeHint.TableData | None
    ) -> None:
        self._ports_in["InputTables"].data = value

    @property
    def OutputTablesRowMask(
        self,
    ) -> PortTypeHint.TablesRowMask | PortTypeHint.TableRowMask | None:
        return self._ports_out["OutputTablesRowMask"].data

    def set_cal_params(
        self, reset: bool = True
    ) -> dict[str, RangeModel | NestedRangeModel] | None:
        input_data: PortTypeHint.TableCollection | PortTypeHint.TableData | None = (
            self._ports_in["InputTables"].data
        )

        filter_by_choices = None
        filter_value_choices = None
        filter_value_depends = None

        if input_data is not None:
            # Handle both TableCollection and single TableData
            if isinstance(input_data, TableCollection):
                # Handle TableCollection (original logic)
                tables = input_data
                if len(tables) > 0:
                    # Find common columns across all tables
                    if len(tables) == 1:
                        # If only one table, use all its field titles
                        filter_by_choices = list(tables)[0].field_titles
                    else:
                        # Find intersection of field titles across all tables
                        common_fields = None
                        for table in tables:
                            table_fields = set(table.field_titles)
                            if common_fields is None:
                                common_fields = table_fields
                            else:
                                common_fields = common_fields.intersection(table_fields)

                        filter_by_choices = (
                            list(common_fields) if common_fields else None
                        )

                    # Reset the filters when it's empty
                    if not self.filters and reset and filter_by_choices:
                        self.filters.append(
                            {
                                "filter_by": filter_by_choices[0],
                                "filter_operation": "eq",
                                "filter_value": None,
                            }
                        )

                    # Get the choices of the filter value for case 'by_uniques'
                    if self.filter_value_choices == "by_uniques":
                        filter_value_depends = "filter_by"
                        filter_value_choices = {}
                        if self.filters:
                            existed_filter_bys = []
                            for filter in self.filters:
                                filter_by = filter["filter_by"]
                                if filter_by and filter_by not in existed_filter_bys:
                                    existed_filter_bys.append(filter_by)
                                    filter_value_choices[filter_by] = set()
                                    for table in tables:
                                        if (
                                            filter_by in table.field_titles
                                            or filter_by in table.columns
                                        ):
                                            filter_value_choices[filter_by] |= set(
                                                table[filter_by].values.tolist()
                                            )
                                    # Convert set to sorted list for RangeModel choices
                                    filter_value_choices[filter_by] = sorted(
                                        list(filter_value_choices[filter_by])
                                    )
                        if not filter_value_choices:
                            filter_value_choices = None
            elif isinstance(input_data, TableData):
                # Handle single TableData
                table = input_data
                filter_by_choices = table.field_titles

                # Reset the filters when it's empty
                if not self.filters and reset and filter_by_choices:
                    self.filters.append(
                        {
                            "filter_by": filter_by_choices[0],
                            "filter_operation": "eq",
                            "filter_value": None,
                        }
                    )

                # Get the choices of the filter value for case 'by_uniques'
                if self.filter_value_choices == "by_uniques":
                    filter_value_depends = "filter_by"
                    filter_value_choices = {}
                    if self.filters:
                        existed_filter_bys = []
                        for filter in self.filters:
                            filter_by = filter["filter_by"]
                            if filter_by and filter_by not in existed_filter_bys:
                                existed_filter_bys.append(filter_by)
                                if (
                                    filter_by in table.field_titles
                                    or filter_by in table.columns
                                ):
                                    filter_value_choices[filter_by] = sorted(
                                        list(set(table[filter_by].values.tolist()))
                                    )
                    if not filter_value_choices:
                        filter_value_choices = None

        filter_ops = list(self.FILTER_OPERATIONS.keys())
        filter_by_range = RangeModel(
            vtype="str",
            title="过滤字段",
            choices=filter_by_choices,
            read_only=self.filter_by_read_only,
            visible=self.fiter_by_visible,
        )
        filter_operation_range = RangeModel(
            vtype="str",
            title="过滤条件",
            choices=filter_ops,
            choices_label=[
                "等于",
                "不等于",
                "大于",
                "小于",
                "大于等于",
                "小于等于",
                "在...中",
                "不在...中",
                "包含",
                "以...开始",
                "以...结束",
            ],
            read_only=self.filter_operation_read_only,
            visible=self.filter_operation_visible,
        )

        filter_value_range = RangeModel(
            vtype=self.filter_value_type or "str",
            title="过滤值",
            choices=filter_value_choices,
            depends_on=filter_value_depends,
        )

        values_range = {
            "filters": NestedRangeModel(
                vtype="list_dict",
                title="条件过滤器",
                fields={
                    "filter_by": filter_by_range,
                    "filter_operation": filter_operation_range,
                    "filter_value": filter_value_range,
                },
                default=self.filters,
                min_max_items=(self.min_items, self.max_items),
            )
        }
        return values_range

    def execute(self) -> PortTypeHint.TableCollection | PortTypeHint.TableData | None:
        input_data: PortTypeHint.TableCollection | PortTypeHint.TableData | None = (
            self._ports_in["InputTables"].data
        )
        module_attributes: PortTypeHint.Attributes | None = self._reserved_ports_in[
            "InputAttributes"
        ].data

        if self.input_attributes_port_required:
            if (
                input_data is None
                or module_attributes is None
                or self.module_attributes_map is None
            ):
                self._ports_out["OutputTables"].data = None
                self._ports_out["OutputTablesRowMask"].data = None
                return None
        else:
            if input_data is None:
                self._ports_out["OutputTables"].data = None
                self._ports_out["OutputTablesRowMask"].data = None
                return None

        if module_attributes:
            for key, value in self.module_attributes_map.items():
                # key is the attribute name of the input module
                # value is the attribute name of the current module
                if key not in module_attributes.keys():
                    raise ValueError(
                        f"The attribute '{key}' is not found in the attributes of the input module '{module_attributes['mname']}'."
                    )
                if isinstance(value, tuple | list):
                    # Check if the base attribute exists (e.g., "filters")
                    if not hasattr(self, value[0]):
                        raise ValueError(
                            f"The attribute '{value[0]}' is not found in the attributes of the current module '{self.mname}'."
                        )

                    # For list[dict] format, check if the dict key exists in the list items
                    if len(value) >= 2:
                        module_value = getattr(self, value[0])
                        if isinstance(module_value, list) and len(module_value) > 0:
                            # Check if the key exists in the first dict item (assume all items have same structure)
                            if (
                                isinstance(module_value[0], dict)
                                and value[1] not in module_value[0]
                            ):
                                raise ValueError(
                                    f"The key '{value[1]}' is not found in the list[dict] attribute '{value[0]}' of the current module '{self.mname}'."
                                )
                elif isinstance(value, str):
                    if value not in self.OutputAttributes.keys():
                        raise ValueError(
                            f"The attribute '{value}' is not found in the attributes of the current module '{self.mname}'."
                        )
                else:
                    raise ValueError(
                        f"The the type of the value of module_attributes_map '{type(value)}' is not a valid type."
                    )

                if isinstance(
                    value, tuple | list
                ):  # The attribute of the module is a list[dict]
                    module_value = getattr(self, value[0])
                    if isinstance(module_value, list):
                        # Handle new list[dict] format: update the specific key in all dict items
                        if len(value) >= 2:
                            dict_key = value[1]
                            for dict_item in module_value:
                                if (
                                    isinstance(dict_item, dict)
                                    and dict_key in dict_item
                                ):
                                    dict_item[dict_key] = module_attributes[key]
                    else:
                        # Handle old dict format (backward compatibility)
                        dict_value = module_value[value[1]]
                        if isinstance(dict_value, list):
                            # Set every value in the list to be the other module's attribute value
                            module_value[value[1]] = [
                                module_attributes[key] for _ in range(len(dict_value))
                            ]
                        else:
                            module_value[value[1]] = module_attributes[key]
                else:
                    setattr(self, value, module_attributes[key])

        # Check if we have any filters
        if not self.filters:
            # No filters, return None
            self._ports_out["OutputTables"].data = None
            self._ports_out["OutputTablesRowMask"].data = None
            return None

        # Handle both TableCollection and single TableData
        if isinstance(input_data, TableCollection):
            # Original logic for TableCollection
            tables = input_data
            output_tables = TableCollection()
            tables_row_mask = {}

            for table in tables:
                # Start with all rows selected
                table_mask = None
                # Apply filters
                for filter_config in self.filters:
                    filter_by = filter_config["filter_by"]

                    # Skip if the table doesn't have this field
                    if (filter_by not in table.field_titles) and (
                        filter_by not in table.columns
                    ):
                        continue

                    # Get the filter function for this condition
                    filter_func = self._get_filter_function(filter_config)
                    # Apply the filter
                    current_mask = table[filter_by].apply(filter_func)
                    # Combine with previous filters (AND operation)
                    if table_mask is None:
                        table_mask = current_mask
                    else:
                        table_mask = table_mask & current_mask
                # If any filters were applied to this table
                if table_mask is not None:
                    filtered_table = table[table_mask]
                    output_tables.add_table(filtered_table)
                    tables_row_mask[table.name] = table_mask.values

            self._ports_out["OutputTables"].data = output_tables
            self._ports_out["OutputTablesRowMask"].data = tables_row_mask
            return output_tables

        elif isinstance(input_data, TableData):
            # New logic for single TableData
            table = input_data

            # Start with all rows selected
            table_mask = None
            # Apply filters
            for filter_config in self.filters:
                filter_by = filter_config["filter_by"]

                # Skip if the table doesn't have this field
                if (filter_by not in table.field_titles) and (
                    filter_by not in table.columns
                ):
                    continue

                # Get the filter function for this condition
                filter_func = self._get_filter_function(filter_config)
                # Apply the filter
                current_mask = table[filter_by].apply(filter_func)
                # Combine with previous filters (AND operation)
                if table_mask is None:
                    table_mask = current_mask
                else:
                    table_mask = table_mask & current_mask

            # If any filters were applied to this table
            if table_mask is not None:
                filtered_table = table[table_mask]
                self._ports_out["OutputTables"].data = filtered_table
                self._ports_out["OutputTablesRowMask"].data = table_mask.values
                return filtered_table
            else:
                # No filters were applied, return None
                self._ports_out["OutputTables"].data = None
                self._ports_out["OutputTablesRowMask"].data = None
                return None


@module_decorator()
class DropDuplicateRows(PipeModule):
    """Drop duplicate rows from a table."""

    InputTable: PortReference[PortTypeHint.TableData]
    OutputTable: PortReference[PortTypeHint.TableData]

    _port_docs = {
        "InputTable": "The table to drop duplicate rows from.",
        "OutputTable": "The table with duplicate rows dropped.",
    }

    def __init__(
        self,
        mname: str = "DropDuplicateRows",
        auto_run: bool = True,
        table: PortTypeHint.TableData | None = None,
        subset: list[str] | None = None,
        keep_empty_strings: bool = True,
        keep_null_values: bool = True,
    ) -> None:
        """Initialize DropDuplicateRows object.

        Parameters
        ----------
        table : TableData
            The table to drop duplicate rows from.

        subset : list[str] | None
            The columns to use to identify duplicate rows.
            If None, all columns will be used.

        keep_empty_strings : bool, default: True
            Whether to keep rows with empty strings in the output.
            If False, rows containing empty strings will be removed.

        keep_null_values : bool, default: True
            Whether to keep rows with None or NaN values in the output.
            If False, rows containing None or NaN values will be removed.

        Ports
        -----
        InputTable: PortTypeHint.TableData
            The table to drop duplicate rows from.

        OutputTable: PortTypeHint.TableData
            The table with duplicate rows dropped.
        """
        super().__init__(mname, auto_run)

        if table is not None:
            self.InputTable = table

        self.subset = subset
        self.keep_empty_strings = keep_empty_strings
        self.keep_null_values = keep_null_values

    def update_ui_schema(self, reset: bool = False) -> dict[str, UIAttributeSchema]:
        table: PortTypeHint.TableData | None = self._ports_in["InputTable"].data

        subset_selections = None
        subset_selections_name = None
        if table is not None:
            subset_selections = table.field_names
            subset_selections_name = table.field_titles

        subset_schema = ArrayAttributeSchema(
            title="去重列名",
            items=StringAttributeSchema(),
            selections=subset_selections,
            selections_name=subset_selections_name,
        )
        subset_schema.widget = "select"

        keep_empty_strings_schema = BooleanAttributeSchema(
            title="保留空字符串",
            default=True,
        )

        keep_null_values_schema = BooleanAttributeSchema(
            title="保留空值(None/NaN)",
            default=True,
        )

        return {
            "subset": subset_schema,
            "keep_empty_strings": keep_empty_strings_schema,
            "keep_null_values": keep_null_values_schema,
        }

    def set_cal_params(self, reset: bool = True) -> dict[str, RangeModel]:
        table: PortTypeHint.TableData | None = self._ports_in["InputTable"].data
        if table is None:
            subset_choices = None
            list_len = None
        else:
            subset_choices = table.field_titles
            list_len = len(subset_choices)

        if not self.subset and reset:
            self.subset = None

        values_range = {
            "subset": RangeModel(
                vtype="list",
                title="去重列名",
                default=self.subset,
                list_type="str",
                required=False,
                choices=subset_choices,
                list_len=list_len,
            ),
            "keep_empty_strings": RangeModel(
                vtype="bool", title="保留空字符串", default=self.keep_empty_strings
            ),
            "keep_null_values": RangeModel(
                vtype="bool", title="保留空值(None/NaN)", default=self.keep_null_values
            ),
        }
        return values_range

    def execute(self) -> PortTypeHint.TableData | None:
        table: PortTypeHint.TableData | None = self._ports_in["InputTable"].data
        if table is None:
            self._ports_out["OutputTable"].data = None
            return None

        column_names = []
        if self.subset:
            tile_to_name = table.title_to_name
            for col in self.subset:
                if col in table.field_titles:
                    column_names.append(tile_to_name[col])
                else:
                    column_names.append(col)
        if len(column_names) == 0:
            column_names = None

        # Drop duplicates first
        table = table.drop_duplicates(subset=column_names, ignore_index=True)

        # Filter out empty strings if requested
        if not self.keep_empty_strings:
            # Create mask for rows that don't contain empty strings
            mask = (table != "").all(axis=1)
            table = table[mask].reset_index(drop=True)

        # Filter out null values if requested
        if not self.keep_null_values:
            # Create mask for rows that don't contain null values
            mask = table.notnull().all(axis=1)
            table = table[mask].reset_index(drop=True)

        self._ports_out["OutputTable"].data = table
        return table


@status_manage
class SingleResultSelector(PipeModule):
    """Select data from SingelReuslt"""

    def __init__(
        self,
        mname: str = "GdimReportTextFilter",
        auto_run: bool = True,
        single_result: PortTypeHint.SingleResult | None = None,
        selected_items: list[str] | str | None = None,
        multiple_select: bool = True,
        merged_name: str | None = None,
        merged_title: str | None = None,
    ) -> None:
        """Initialize TextLibraryResultFilter object.

        Parameters
        ----------
        single_result : SingleResult
            The single result to filter.

        selected_items : list[str] | str, default: None
            The name or title of the items to select.
            If multiple_select is False, the type of seleted_items is str.

        multiple_select: bool, default: True
            Whether to allow multiple selection.

        merged_name: str | None, default: None
            If a string is given, all the selected items will be merged into one UnitResult using the string as name.
            All the slected items will be merged into a list. The Unit and description of the first selected item will be used.
            If multiple_select is False, only the name will be changed.

        merged_title: str | None, default: None
            Only available when merged_name is not None. This string will be used as the title of the merged UnitResult.
            If None, title will use merged_name.
        """
        super().__init__(mname=mname, auto_run=auto_run)

        self._ports_in = {
            "InputSingleResult": Port(
                PortType.SingleResult,
                data=single_result,
                pdoc="The single result to filter.",
            )
        }
        self._ports_out = {
            "OutputSingleResult": Port(
                PortType.SingleResult, pdoc="The selected single result."
            )
        }

        self.selected_items = selected_items
        self.multiple_select = multiple_select
        self.merged_name = merged_name
        self.merged_title = merged_title

    @property
    def InputSingleResult(self) -> None:
        raise AttributeError("Property 'InputSingleResult' is read-only.")

    @InputSingleResult.setter
    def InputSingleResult(self, value: PortTypeHint.SingleResult | None) -> None:
        self._ports_in["InputSingleResult"].data = value

    @property
    def OutputSingleResult(self) -> PortTypeHint.SingleResult | None:
        return self._ports_out["OutputSingleResult"].data

    def set_cal_params(self, reset: bool = True) -> dict[str, RangeModel]:
        single_result: PortTypeHint.SingleResult | None = self._ports_in[
            "InputSingleResult"
        ].data

        if single_result is None:
            selected_items_choices = None
        else:
            selected_items_choices = single_result.titles
            if not self.selected_items and reset:
                if self.multiple_select:
                    self.selected_items = selected_items_choices
                else:
                    self.selected_items = [selected_items_choices[0]]

        if self.multiple_select:
            list_len = None
        else:
            list_len = 1

        values_range = {
            "selected_items": RangeModel(
                vtype="list",
                list_len=list_len,
                title="选择的条目",
                default=self.selected_items,
                choices=selected_items_choices,
            )
        }
        return values_range

    def execute(self) -> PortTypeHint.SingleResult | None:
        single_result: PortTypeHint.SingleResult | None = self._ports_in[
            "InputSingleResult"
        ].data
        if single_result is None or self.selected_items is None:
            self._ports_out["OutputSingleResult"].data = None
            return None

        unit_result_list = []
        if isinstance(self.selected_items, str):
            unit_result_list.append(single_result[self.selected_items])
        else:
            for selected_item in self.selected_items:
                unit_result_list.append(single_result[selected_item])

        if self.merged_name:
            if self.multiple_select:
                value = []
                for unit_result in unit_result_list:
                    value.append(unit_result.value)
                output = SingleResult(
                    [
                        UnitResult(
                            name=self.merged_name,
                            title=self.merged_title or self.merged_name,
                            unit=unit_result_list[0].unit,
                            value=value,
                            description=unit_result_list[0].description,
                        )
                    ]
                )
            else:
                output = SingleResult(
                    [
                        UnitResult(
                            name=self.merged_name,
                            title=self.merged_title or self.merged_name,
                            unit=unit_result_list[0].unit,
                            value=unit_result_list[0].value,
                            description=unit_result_list[0].description,
                        )
                    ]
                )
        else:
            output = SingleResult(unit_result_list)

        self._ports_out["OutputSingleResult"].data = output


@status_manage
class GdimReportTemplateSelector(PipeModule):
    """Select the data from SingelReuslt from GdimAppReportTextReader with `read_document=True`.
    The output will be the template_path, template_name and output_name which are the attributes value of DocPrinter.
    """

    def __init__(
        self,
        mname: str = "GdimReportTemplateSelector",
        auto_run: bool = True,
        single_result: PortTypeHint.SingleResult | None = None,
        selected_items: list[str] | str | None = None,
        multiple_select: bool = True,
    ) -> None:
        """Initialize GdimReportTemplateSelector object.

        Parameters
        ----------
        single_result : SingleResult, default: None
            The single result to select.

        selected_items : list[str] | str | None, default: None
            The selected items to select.
            If multiple_select is True, the type is list[str].
            If multiple_select is False, the type is list[str] or str.

        multiple_select : bool, default: True
            Whether to allow multiple selection.
        """
        super().__init__(mname, auto_run)

        self._ports_in = {
            "InputSingleResult": Port(
                PortType.SingleResult,
                data=single_result,
                pdoc="The single result to select.",
            )
        }
        self._ports_out = {
            "OutputDocPrinterAttributes": Port(
                PortType.Attributes,
                pdoc="The attributes (template_path, template_name, output_name) of the DocPrinter.",
            ),
            "OutputSingleResultDocs": Port(
                PortType.SingleResult,
                pdoc="The single result of the selected docs not template.",
            ),
        }

        self.selected_items = selected_items
        self.multiple_select = multiple_select

    @property
    def InputSingleResult(self) -> None:
        raise AttributeError("Property 'InputSingleResult' is read-only.")

    @InputSingleResult.setter
    def InputSingleResult(self, value: PortTypeHint.SingleResult | None) -> None:
        self._ports_in["InputSingleResult"].data = value

    @property
    def OutputDocPrinterAttributes(self) -> PortTypeHint.Attributes | None:
        return self._ports_out["OutputDocPrinterAttributes"].data

    @property
    def OutputSingleResultDocs(self) -> PortTypeHint.SingleResult | None:
        return self._ports_out["OutputSingleResultDocs"].data

    def set_cal_params(self, reset: bool = True) -> dict[str, RangeModel]:
        single_result: PortTypeHint.SingleResult | None = self._ports_in[
            "InputSingleResult"
        ].data

        if single_result is None:
            selected_items_choices = None
        else:
            selected_items_choices = single_result.titles
            if not self.selected_items and reset:
                if self.multiple_select:
                    self.selected_items = selected_items_choices
                else:
                    self.selected_items = [selected_items_choices[0]]

        if self.multiple_select:
            list_len = None
        else:
            list_len = 1

        values_range = {
            "selected_items": RangeModel(
                vtype="list",
                list_len=list_len,
                title="选择的条目",
                default=self.selected_items,
                choices=selected_items_choices,
            )
        }
        return values_range

    def execute(self) -> PortTypeHint.Attributes | None:
        single_result: PortTypeHint.SingleResult | None = self._ports_in[
            "InputSingleResult"
        ].data
        if single_result is None or self.selected_items is None:
            self._ports_out["OutputDocPrinterAttributes"].data = None
            self._ports_out["OutputSingleResultDocs"].data = None
            return None

        if isinstance(self.selected_items, str):
            selected_items = [self.selected_items]
        else:
            selected_items = self.selected_items

        doc_unit_results = []
        output = {"template_path": None, "template_name": [], "output_name": []}
        for selected_item in selected_items:
            unit_result: UnitResult = single_result[selected_item]
            if unit_result.description == DataTypes.FilePath:
                doc_unit_results.append(unit_result)
            elif unit_result.description == DataTypes.TemplatePath:
                template_file = Path(unit_result.value)
                output_name = unit_result.name.replace("-", "_")
                output_name = output_name.replace("模板", "")
                output["template_path"] = str(template_file.parent)
                output["template_name"].append(template_file.name)
                output["output_name"].append(output_name + ".docx")

        self._ports_out["OutputDocPrinterAttributes"].data = output
        self._ports_out["OutputSingleResultDocs"].data = SingleResult(
            result=doc_unit_results
        )
        return output


@status_manage
class GdimReportTextSelector(PipeModule):
    """Select the data from SingelReuslt from GdimAppReportTextReader with `read_document=False`."""

    def __init__(
        self,
        mname: str = "GdimReportTextSelector",
        auto_run: bool = True,
        single_result: PortTypeHint.SingleResult | None = None,
        selected_items: list[dict[str, list | str]] | None = None,
        min_items: int = 0,
        max_items: int | None = None,
        text_multiple_select: bool = True,
        group_title_read_only: bool = False,
    ) -> None:
        """Initialize GdimReportTextSelector object.

        Parameters
        ----------
        single_result : SingleResult, default: None
            The single result to select.

        selected_items : list[dict[str, list | str]] | None, default: None
            The selected items to select.
            Two keys are available -
                "group_title" - The selected group title. Always single select, the type is str.
                "text_indexes" - The selected text indexes. Single select or multiple select, the type is int or list[int].

        min_items : int, default: 0
            The minimum number of selected items.

        max_items : int | None, default: None
            The maximum number of selected items.
            If None, no limit.

        text_multiple_select : bool, default: True
            Whether to allow multiple selection for text_indexes in each group.
            Note: group_title is always single select.

        group_title_read_only : bool, default: False
            Whether to make the group_title read only.
        """
        super().__init__(mname, auto_run)

        self._ports_in = {
            "InputSingleResult": Port(
                PortType.SingleResult,
                data=single_result,
                pdoc="The single result to select.",
            )
        }
        self._ports_out = {
            "OutputSingleResult": Port(
                PortType.SingleResult, pdoc="The selected single result."
            )
        }

        # selected_items example - [{"group_title": str, "text_indexes": list[int]}, {"group_title": str, "text_indexes": list[int]}, ...]
        self.selected_items = selected_items or []
        self.min_items = min_items
        self.max_items = max_items
        self.text_multiple_select = text_multiple_select
        self.group_title_read_only = group_title_read_only

    @property
    def InputSingleResult(self) -> None:
        raise AttributeError("Property 'InputSingleResult' is read-only.")

    @InputSingleResult.setter
    def InputSingleResult(self, value: PortTypeHint.SingleResult | None) -> None:
        self._ports_in["InputSingleResult"].data = value

    @property
    def OutputSingleResult(self) -> PortTypeHint.SingleResult | None:
        return self._ports_out["OutputSingleResult"].data

    def set_cal_params(self, reset: bool = True) -> dict[str, RangeModel]:
        single_result: PortTypeHint.SingleResult | None = self._ports_in[
            "InputSingleResult"
        ].data

        if single_result is None or not self.selected_items:
            group_title_choices = None
            text_indexes_choices = None
            text_indexes_choices_label = None
        else:
            # The value of the single result must be list[str] - Text content, cannot be FilePath content
            if single_result:
                if single_result.descriptions[0] != DataTypes.Text:
                    raise ValueError(
                        f"The value of the data from port 'InputSingleResult' must be list[str] which means only text content from the ReportText is available!"
                    )

            group_title_choices = single_result.titles
            text_indexes_choices = {}
            text_indexes_choices_label = {}

            for title in group_title_choices:
                text_indexes_choices[title] = list(
                    range(len(single_result[title].value))
                )
                text_indexes_choices_label[title] = [
                    text for text in single_result[title].value
                ]

            if not self.selected_items and reset:
                group_title = group_title_choices[0]
                self.selected_items = [
                    {
                        "group_title": group_title,
                        "text_indexes": single_result[group_title].value,
                    }
                ]

            if self.selected_items and reset:
                for idx, item in enumerate(self.selected_items):
                    if item["group_title"] and not item["text_indexes"]:
                        self.selected_items[idx]["text_indexes"] = list(
                            range(len(single_result[item["group_title"]].value))
                        )

        if self.text_multiple_select:
            text_list_len = None
        else:
            text_list_len = 1

        items_range = NestedRangeModel(
            vtype="list_dict",
            title="预设文本选择",
            fields={
                "group_title": RangeModel(
                    vtype="str",
                    title="组标题",
                    choices=group_title_choices,
                    read_only=self.group_title_read_only,
                ),
                "text_indexes": RangeModel(
                    vtype="list",
                    list_len=text_list_len,
                    title="选择预设文本",
                    choices=text_indexes_choices,
                    choices_label=text_indexes_choices_label,
                    depends_on=["group_title"],
                ),
            },
            default=self.selected_items,
            min_max_items=(self.min_items, self.max_items),
        )

        return {"selected_items": items_range}

    def execute(self) -> PortTypeHint.SingleResult | None:
        single_result_input: PortTypeHint.SingleResult | None = self._ports_in[
            "InputSingleResult"
        ].data

        if single_result_input is None or not self.selected_items:
            self._ports_out["OutputSingleResult"].data = None
            return None

        # The value of the single result must be list[str] - Text content, cannot be FilePath content
        if single_result_input:
            if single_result_input.descriptions[0] != DataTypes.Text:
                raise ValueError(
                    f"The value of the data from port 'InputSingleResult' must be list[str] which means only text content from the ReportText is available!"
                )

        unit_result_list = []
        for index, item in enumerate(self.selected_items):
            group_title = item["group_title"]
            text_indexes = item["text_indexes"]
            selected_result = single_result_input[group_title]

            value = [selected_result.value[idx] for idx in text_indexes]
            unit_result = UnitResult(
                name=group_title + str(index),
                title=group_title,
                value=value,
                unit=selected_result.unit,
                description=selected_result.description,
            )
            unit_result_list.append(unit_result)

        selected_single_result = SingleResult(unit_result_list)

        self._ports_out["OutputSingleResult"].data = selected_single_result
        return selected_single_result


@status_manage
class ResultsDictSelector(PipeModule):
    """Select the value from ResultsDict or MultiResultsDict.

    This module can extract data from both ResultsDict (single pipeline run) and
    MultiResultsDict (multiple pipeline runs). It supports selecting:
    - Output port data (module_name@port_name)
    - Pipeline attributes (pipeline@attr_name)
    - Module attributes (module_name#attr_name)
    """

    def __init__(
        self,
        mname: str | None = "ResultsDictSelector",
        auto_run: bool = True,
        results_dict: (
            PortTypeHint.ResultsDict | PortTypeHint.MultiResultsDict | None
        ) = None,
        pipeline_name: str | None = None,
        name: str | None = None,
        module_name: str | None = None,
        run_id: int | None = None,
    ) -> None:
        """Initialize ResultsDictSelector object.

        Parameters
        ----------
        results_dict : ResultsDict | MultiResultsDict, optional
            The results dictionary to select from.

        pipeline_name : str, optional
            The name of the pipeline to select the result from.
            Only used when the results_dict is a MultiResultsDict with multiple pipelines.

        name : str, optional
            The name of the data item to select:
            - For output ports: the port name
            - For pipeline attributes: the attribute name
            - For module attributes: the attribute name
            If None, returns the entire ResultsDict for the specified pipeline.

        module_name : str, optional
            The module name. Required when selecting output ports or module attributes.
            Not used for pipeline attributes.

        run_id : int, optional
            The run number to select from (1, 2, 3, ...). Only used with MultiResultsDict.
            - Use positive integers (1, 2, 3, ...) to select specific runs
            - Use -1 to select the latest run
            - If None, returns all results from all runs combined

        Note
        ----
        If run_id is None and InputResultsDict is a MultiResultsDict, the output will always be a list.
        """
        super().__init__(mname=mname, auto_run=auto_run)

        self._ports_in = {
            "InputResultsDict": Port(
                [PortType.ResultsDict, PortType.MultiResultsDict],
                data=results_dict,
                pdoc="The results dictionary to select from.",
            )
        }
        self._ports_out = {
            "OutputResult": Port(
                PortType.General,
                pdoc="The selected result - can be any data type stored in ResultsDict.",
            )
        }

        self.pipeline_name = pipeline_name
        self.name = name
        self.module_name = module_name
        self.run_id = run_id

    @property
    def InputResultsDict(self) -> None:
        raise AttributeError("Property 'InputResultsDict' is write-only.")

    @InputResultsDict.setter
    def InputResultsDict(self, value: PortTypeHint.ResultsDict | None) -> None:
        self._ports_in["InputResultsDict"].data = value

    @property
    def OutputResult(self) -> Any:
        return self._ports_out["OutputResult"].data

    def set_cal_params(self, reset: bool = True) -> None:
        return None

    def execute(self) -> Any:
        input_dict: PortTypeHint.ResultsDict | PortTypeHint.MultiResultsDict | None = (
            self._ports_in["InputResultsDict"].data
        )

        if input_dict is None:
            self._ports_out["OutputResult"].data = None
            return None

        # Get the target ResultsDict to work with
        target_results_dict = None

        # Handle MultiResultsDict
        if isinstance(input_dict, MultiResultsDict):
            if not self.pipeline_name:
                warnings.warn("pipeline_name is required when using MultiResultsDict")
                self._ports_out["OutputResult"].data = None
                return None

            if self.pipeline_name not in input_dict.pipelines:
                warnings.warn(
                    f"Pipeline '{self.pipeline_name}' not found in MultiResultsDict. Available pipelines: {list(input_dict.pipelines.keys())}"
                )
                self._ports_out["OutputResult"].data = None
                return None

            string_run_ids = list(input_dict.pipelines[self.pipeline_name].keys())
            num_runs = len(string_run_ids)

            if self.run_id is not None:
                if self.run_id == -1:
                    # Use latest run
                    target_results_dict = input_dict.get_latest_result(
                        self.pipeline_name
                    )
                    if target_results_dict is None:
                        warnings.warn(
                            f"No runs found for pipeline '{self.pipeline_name}'"
                        )
                        self._ports_out["OutputResult"].data = None
                        return None
                elif 1 <= self.run_id <= num_runs:
                    # Use specific run (convert integer to string format)
                    string_run_id = f"run_{self.run_id}"
                    target_results_dict = input_dict.get_result(
                        self.pipeline_name, string_run_id
                    )
                    if target_results_dict is None:
                        warnings.warn(
                            f"Run {self.run_id} not found for pipeline '{self.pipeline_name}'. Available runs: 1-{num_runs}"
                        )
                        self._ports_out["OutputResult"].data = None
                        return None
                else:
                    warnings.warn(
                        f"Invalid run_id {self.run_id}. Use 1-{num_runs} for specific runs or -1 for latest run."
                    )
                    self._ports_out["OutputResult"].data = None
                    return None
            else:
                # No specific run_id, combine all runs
                all_data = {}
                all_source_info = {}
                for string_run_id in string_run_ids:
                    run_results = input_dict.get_result(
                        self.pipeline_name, string_run_id
                    )
                    if run_results:
                        # Use the string_run_id directly as it's already unique within the pipeline
                        for key, value in run_results.data.items():
                            combined_key = f"{string_run_id}@{key}"
                            all_data[combined_key] = value
                            all_source_info[combined_key] = run_results.source_info.get(
                                key, "unknown"
                            )

                target_results_dict = ResultsDict(
                    data=all_data, source_info=all_source_info
                )
        else:
            # Handle single ResultsDict
            target_results_dict = input_dict

        # If no specific name requested, return the entire ResultsDict
        if self.name is None:
            self._ports_out["OutputResult"].data = target_results_dict
            return target_results_dict

        # Build the key to look for based on the data type
        selected_key = None

        # Helper function to check if a key matches the pattern with optional run_id prefix
        def key_matches_pattern(key: str, target_pattern: str) -> bool:
            """Check if key matches target pattern, accounting for optional run_id@ prefix"""
            if key == target_pattern:
                return True
            # Check if key has run_id@ prefix and matches pattern after removing it
            if "@" in key:
                parts = key.split("@", 1)  # Split on first @ only
                if (
                    len(parts) == 2
                    and parts[0].startswith("run_")
                    and parts[1] == target_pattern
                ):
                    return True
            return False

        # Try different key formats based on module_name and name
        # Collect all matching keys first, then decide which one(s) to return
        matching_keys_by_pattern = []

        if self.module_name:
            # Try output port format: module_name@port_name
            port_key = f"{self.module_name}@{self.name}"
            for key in target_results_dict.keys():
                if key_matches_pattern(key, port_key):
                    matching_keys_by_pattern.append(key)

            if not matching_keys_by_pattern:
                # Try module attribute format: module_name#attr_name
                attr_key = f"{self.module_name}#{self.name}"
                for key in target_results_dict.keys():
                    if key_matches_pattern(key, attr_key):
                        matching_keys_by_pattern.append(key)
        else:
            # Try pipeline attribute format: pipeline@attr_name
            pipeline_key = f"pipeline@{self.name}"
            for key in target_results_dict.keys():
                if key_matches_pattern(key, pipeline_key):
                    matching_keys_by_pattern.append(key)

            if not matching_keys_by_pattern:
                # Try exact match
                for key in target_results_dict.keys():
                    if key_matches_pattern(key, self.name):
                        matching_keys_by_pattern.append(key)

        # Handle multiple matches from combined runs
        if (
            len(matching_keys_by_pattern) >= 1
            and self.run_id is None
            and isinstance(input_dict, MultiResultsDict)
        ):
            # Return all matching values as a list，ignore the None result
            matching_results = [
                target_results_dict[key]
                for key in matching_keys_by_pattern
                if target_results_dict[key] is not None
            ]
            if len(matching_results) == 0:
                # warnings.warn(f"No results found for name='{self.name}', module_name='{self.module_name}'")
                self._ports_out["OutputResult"].data = None
                return None
            self._ports_out["OutputResult"].data = matching_results
            return matching_results

        if len(matching_keys_by_pattern) == 1:
            selected_key = matching_keys_by_pattern[0]

        # If still no match, try fuzzy matching
        """
        if selected_key is None:
            # Try to find keys that contain the name
            matching_keys = []
            for key in target_results_dict.keys():
                if self.name in key:
                    matching_keys.append(key)
            
            if len(matching_keys) == 1:
                selected_key = matching_keys[0]
            elif len(matching_keys) > 1:
                # If module_name is specified, prefer keys that contain it
                if self.module_name:
                    module_matches = [key for key in matching_keys if self.module_name in key]
                    if len(module_matches) == 1:
                        selected_key = module_matches[0]
                    elif len(module_matches) > 1:
                        # Prefer exact module match (accounting for run_id prefix)
                        exact_matches = []
                        for key in module_matches:
                            # Check direct patterns
                            if key.startswith(f"{self.module_name}@") or key.startswith(f"{self.module_name}#"):
                                exact_matches.append(key)
                            # Check patterns with run_id prefix
                            elif "@" in key:
                                parts = key.split("@", 1)
                                if len(parts) == 2 and parts[0].startswith("run_"):
                                    remaining = parts[1]
                                    if remaining.startswith(f"{self.module_name}@") or remaining.startswith(f"{self.module_name}#"):
                                        exact_matches.append(key)
                        
                        if exact_matches:
                            selected_key = exact_matches[0]
                        else:
                            selected_key = module_matches[0]
                    else:
                        selected_key = matching_keys[0]
                else:
                    selected_key = matching_keys[0]
        """

        if selected_key is None:
            available_keys = list(target_results_dict.keys())
            warnings.warn(
                f"Could not find data for name='{self.name}', module_name='{self.module_name}'. Available keys: {available_keys}"
            )
            self._ports_out["OutputResult"].data = None
            return None

        selected_result = target_results_dict[selected_key]
        self._ports_out["OutputResult"].data = selected_result
        return selected_result
